package com.example.toy.core.fs;

import com.example.toy.core.conf.ToyTableConfig;
import com.example.toy.core.dataset.BigCollection;
import com.example.toy.core.engine.EngineContext;
import com.example.toy.core.engine.GlobalEngineContext;
import com.example.toy.core.table.TableIdentifier;
import com.example.toy.core.utils.ToyPathDefine;
import lombok.SneakyThrows;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * FsOfflineToyTableFileView
 *
 * @author shenbing
 * @date 2021/12/24
 */
public class FsOfflineToyTableFileView implements ToyTableFileView {

  private final TableIdentifier tableIdentifier;
  private final Path basePath;
  private final Path dataDir;
  private final FileSystem fs;

  @SneakyThrows
  public FsOfflineToyTableFileView(TableIdentifier tableIdentifier, Path basePath, Path dataDir) {
    this.tableIdentifier = tableIdentifier;
    this.basePath = basePath;
    this.dataDir = dataDir;
    this.fs = this.basePath.getFileSystem(new Configuration());
  }

  public FsOfflineToyTableFileView(TableIdentifier tableIdentifier, Path basePath) {
    this(tableIdentifier, basePath, basePath);
  }

  @SneakyThrows
  @Override
  public FileStatus getToyMetaDir() {
    Path metaDirPath = ToyPathDefine.getMetaDirPath(basePath);
    return fs.exists(metaDirPath) ? null : fs.getFileStatus(metaDirPath);
  }

  @SneakyThrows
  @Override
  public FileStatus getToyConfigFile() {
    Path configFilePath = ToyPathDefine.getTableConfigFilePath(basePath);
    return fs.exists(configFilePath) ? null : fs.getFileStatus(configFilePath);
  }

  @SneakyThrows
  @Override
  public FileStatus getToyCommitDir() {
    Path commitsDirPath = ToyPathDefine.getCommitDirPath(basePath);
    return fs.exists(commitsDirPath) ? null : fs.getFileStatus(commitsDirPath);
  }

  @SneakyThrows
  @Override
  public List<FileStatus> getToyCommitFiles() {
    List<FileStatus> list = new ArrayList<>();
    Path commitsDirPath = ToyPathDefine.getCommitDirPath(basePath);
    if (fs.exists(commitsDirPath)) {
      list.addAll(Arrays.asList(fs.getFileStatus(commitsDirPath)));
    }
    return list;
  }

  /** @return */
  @SneakyThrows
  @Override
  public List<FileStatus> getPartitionDirs() {
    Path configFilePath = ToyPathDefine.getTableConfigFilePath(this.basePath);
    ToyTableConfig tableConfig = ToyTableConfig.fromCacheOrPath(configFilePath);
    List<String> partitionFields = tableConfig.getPartitionFields();

    List<FileStatus> list = new ArrayList<>();
    if (!partitionFields.isEmpty() && fs.exists(this.dataDir)) {
      EngineContext engineContext = GlobalEngineContext.get();
      BigCollection<FileStatus> bigCollection =
          engineContext.create(fs.getFileStatus(this.dataDir));
      while (partitionFields.size() > 0) {
        partitionFields.remove(0);
        bigCollection =
            bigCollection.flatMap(
                p -> {
                  FileSystem f = p.getPath().getFileSystem(new Configuration());
                  FileStatus[] fileStatuses = f.listStatus(p.getPath());
                  return Arrays.asList(fileStatuses).iterator();
                });
      }
      list.addAll(bigCollection.collect());
    }
    return list;
  }
}
